
package cn.edu.thu.laud.lasql.index;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.List;
import java.util.Map;
import java.util.Set;

import org.apache.cassandra.config.ColumnDefinition;
import org.apache.cassandra.config.ConfigurationException;
import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.DecoratedKey;
import org.apache.cassandra.db.IColumn;
import org.apache.cassandra.db.index.PerColumnSecondaryIndex;
import org.apache.cassandra.db.index.SecondaryIndexSearcher;
import org.apache.cassandra.db.marshal.AbstractType;
import org.apache.cassandra.db.marshal.BytesType;
import org.apache.cassandra.db.marshal.LocalByPartionerType;
import org.apache.cassandra.dht.ByteOrderedPartitioner;
import org.apache.cassandra.dht.IPartitioner;
import org.apache.cassandra.dht.OrderPreservingPartitioner;
import org.apache.cassandra.service.StorageService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import cn.edu.thu.laud.lasql.index.iface.LaUDSecondaryIndex;


/**
 * Implements a secondary index for a column family using a second column family
 * in which the row keys are indexed values, and column names are base row keys.
 */
public class LaUDIndex extends PerColumnSecondaryIndex
{

	public static final String IMPLEMENTION_CLASS="implemention_class";
	private static final Logger logger = LoggerFactory.getLogger(LaUDIndex.class);
	//   private ColumnFamilyStore indexCfs;
	private LaUDSecondaryIndex index;
	
	private String indexName;
	private ByteBuffer columnName;
	
	private String columnFamily;
	private String keyspace;
	public LaUDIndex()
	{

	}

	public void init()
	{
		assert baseCfs != null && columnDefs != null;

		ColumnDefinition columnDef = columnDefs.iterator().next();
		columnName=columnDef.name;
		indexName=columnDef.getIndexName();
		columnFamily=baseCfs.columnFamily;
		keyspace=baseCfs.table.name;
		Map<String, String>optionsMap=columnDef.getIndexOptions();
		String implemention_class_name=optionsMap.get(IMPLEMENTION_CLASS);
		assert implemention_class_name!=null;
		
		try
        {
            index = (LaUDSecondaryIndex) Class.forName(implemention_class_name).newInstance();
        }
        catch (Exception e)
        {
            throw new RuntimeException(e);
        }
		
		index.init(keyspace,columnFamily,columnName,indexName,optionsMap);
	}

	/**
	 * @deprecated this method is copy from keysIndex 
	 * <br> infect ,we needn't this method because this method in keysIndex just is called when the cf'name contains "." which  cassandra consider it a hidden cf 
	 * @return
	 */
	public static AbstractType<?> indexComparator()
	{
		IPartitioner rowPartitioner = StorageService.getPartitioner();
		return (rowPartitioner instanceof OrderPreservingPartitioner || rowPartitioner instanceof ByteOrderedPartitioner)
				? BytesType.instance
						: new LocalByPartionerType(StorageService.getPartitioner());
	}
	

	public void deleteColumn(DecoratedKey<?> valueKey, ByteBuffer rowKey, IColumn column)
	{
		try {
			index.deleteColumn(indexName, valueKey.key, rowKey, columnName, columnFamily,keyspace);
		} catch (IOException e) {
			logger.warn("delete column from laudindex failed:"+e.getMessage());
			//e.printStackTrace();
		}
	}

	public void insertColumn(DecoratedKey<?> valueKey, ByteBuffer rowKey, IColumn column)
	{
		try {
			index.insertColumn(indexName, valueKey.key, rowKey, columnName, columnFamily,keyspace);
		} catch (IOException e) {
			logger.warn("insert column from laudindex failed:"+e.getMessage());
			//e.printStackTrace();
		}
	}

	public void updateColumn(DecoratedKey<?> valueKey, ByteBuffer rowKey, IColumn col)
	{
		try {
			index.updateColumn(indexName, valueKey.key, rowKey, columnName, columnFamily,keyspace);
		} catch (IOException e) {
			logger.warn("insert column from laudindex failed:"+e.getMessage());
			//e.printStackTrace();
		}
	}

	public void removeIndex(ByteBuffer columnName) throws IOException
	{
		index.removeIndex(indexName, columnName, columnFamily,keyspace);
	}

	public void forceBlockingFlush() throws IOException
	{
		index.forceBlockingFlush();
	}

	public void invalidate()
	{
		try {
			index.removeIndex(indexName, columnName, columnFamily, keyspace);
		} catch (IOException e) {
			logger.warn("remove column from laudindex failed:"+e.getMessage());
		}
	}

	public void truncate(long truncatedAt)
	{
		index.truncate(truncatedAt);
	}


	public SecondaryIndexSearcher createSecondaryIndexSearcher(Set<ByteBuffer> columns)
	{
		return new LaUDSearcher(baseCfs.indexManager, columns);
	}

	public String getIndexName()
	{
		return indexName;
	}

	public void validateOptions() throws ConfigurationException
	{
		assert  columnDefs != null;

		ColumnDefinition columnDef = columnDefs.iterator().next();
		columnName=columnDef.name;
		indexName=columnDef.getIndexName();
		Map<String, String>optionsMap=columnDef.getIndexOptions();
		if(optionsMap.get(IMPLEMENTION_CLASS)==null)
			throw new ConfigurationException(IMPLEMENTION_CLASS+" is required.");
	}

	public long getLiveSize()
	{
		return index.getLiveSize();
	}
/**
 * laUD Index doesn't support this method.
 */
	@Override
	public ColumnFamilyStore getIndexCfs() {
		logger.warn("program call this unsupported method........");
		return null;
	}
	
	public Object getLaUDIndexCfs() {
		return index.getIndexFiles();
	}
	public LaUDSecondaryIndex getIndex(){
		return this.index;
	}

	public List<ByteBuffer> search(ByteBuffer value,boolean reversed,int limit) {
		List<ByteBuffer>  keys=index.search(keyspace, columnFamily, indexName, columnName, value,reversed,limit,columnDefs.iterator().next().getIndexOptions());
		return keys;
		
		//return rows;
	}
	
}
